package com.dis.demo;

import org.apache.zookeeper.*;
import org.apache.zookeeper.data.Stat;

import java.util.ArrayList;
import java.util.Collections;
import java.util.List;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.Lock;

/**
 * 软谋教育Java VIP课程
 * <pre>
 *    ZK分布式锁实现
 * </pre>
 *
 * @author gerry
 * @date 2018-07-05
 */
public class DistributeLock implements Watcher, Lock {

    // 定义锁的名称属性
    private String lockName;
    private final static String ROOT_LOCK = "/locks";
    // 定义超时时间
    private int sessionTimeout = 30000;
    // 定义存储当前锁的变量
    String current_lock = null;
    // 定义等待的锁
    String wait_lock = null;
    // 创建Zookeeper句柄
    ZooKeeper zk = null;
    // 阻塞线程
    CountDownLatch latch = null;

    public DistributeLock(String url, String lockName) {
        this.lockName = lockName;

        try {
            // 建立与Zookeeper的会话
            zk = new ZooKeeper(url, sessionTimeout , this);
            // 判断根节点是存在
            Stat stat = zk.exists(ROOT_LOCK, false);

            if (stat == null) {
                zk.create(ROOT_LOCK,new byte[0],ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
            }
        } catch (Exception e) {
            e.printStackTrace();
        }
    }
    /**
     * 加锁
     */
    public void lock() {
        if (tryLock()) {
            System.out.println(Thread.currentThread().getName()+"->获取锁是:"+current_lock);
            return;
        } else {
            try {
                waitForLock();
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
    }

    /**
     * 等待锁
     */
    private boolean waitForLock() throws InterruptedException {
        try {
            Stat stat = zk.exists(RLOOOT_CK + "/" + wait_lock, true);

            if (stat != null) {
                this.latch = new CountDownLatch(1);
                this.latch.await(sessionTimeout, TimeUnit.MILLISECONDS);
                this.latch = null;
                System.out.println(Thread.currentThread().getName() + " 等到了锁->"+wait_lock);

            }
        } catch (KeeperException e) {
            e.printStackTrace();
        }

        return true;
    }

    /**
     * 尝试获取锁的方法
     * @return
     */
    public boolean tryLock() {
        String splitStr = "_lock_";
        // 获取当前锁
        try {
            Stat stat = zk.exists(ROOT_LOCK + "/" + lockName + splitStr, false);
            if (stat == null) {
                current_lock = zk.create(ROOT_LOCK + "/" + lockName + splitStr, new byte[0], ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL_SEQUENTIAL);
            }

            // 获取当前根节点下面所有子节点
            List<String> children = zk.getChildren(ROOT_LOCK, false);
            // 定义一个集合存储子节点
            List<String> lockObjects = new ArrayList<>();

            for (String node : children) {
                // /locks/test_lock_00010121
                String _node = node.split(splitStr)[0];
                if (_node.equals(lockName)) {
                    lockObjects.add(node);
                }
            }
            // 按照从小到大顺序排序
            Collections.sort(lockObjects);

            // 判断当前锁是否为最小序号
            if (current_lock.equals(ROOT_LOCK + "/" +lockObjects.get(0))) {
                System.out.println(Thread.currentThread().getName()+"获得锁->"+current_lock);
                return true;
            }
            System.out.println(lockObjects);
            // 如果不是最小节点，找到它前一个节点
            String currentNode = current_lock.substring(current_lock.lastIndexOf("/")+1);
            wait_lock = lockObjects.get(Collections.binarySearch(lockObjects, currentNode) -1);
        } catch (KeeperException e) {
            e.printStackTrace();
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        return false;
    }

    /**
     * 释放锁的方法
     */
    public void unlock() {
        try {
            System.out.println(Thread.currentThread().getName() +"释放锁 " + current_lock);
            if (zk.exists(current_lock, false) != null) {
                zk.delete(current_lock, -1);

                current_lock = null;
                zk.close();
            }
        } catch (InterruptedException e) {
            e.printStackTrace();
        } catch (KeeperException e) {
            e.printStackTrace();
        }
    }

    @Override
    public void process(WatchedEvent watchedEvent) {
        if (latch != null) {
            latch.countDown();
        }
    }

    @Override
    public boolean tryLock(long time, TimeUnit unit) throws InterruptedException {
        return false;
    }

    @Override
    public Condition newCondition() {
        return null;
    }

    @Override
    public void lockInterruptibly() throws InterruptedException {

    }
}
